package pers.mihao.quickstream.common.task.scheduler.isolated;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import pers.mihao.quickstream.common.task.constans.TaskType;
import pers.mihao.quickstream.common.task.domain.runner.AutomaticTaskRunner;
import pers.mihao.quickstream.common.task.dto.AutomaticTaskDTO;
import pers.mihao.quickstream.common.task.scheduler.TaskInvokeRes;
import pers.mihao.quickstream.common.util.AssertUtil;

import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * 同步调度-任务执行会被阻塞
 * @author mihao
 * @since 2024/11/20
 */
@Slf4j
public class TaskCallable implements Runnable {
    AutomaticTaskDTO taskDTO;

    CompletableFuture<TaskInvokeRes> future;

    public TaskCallable(AutomaticTaskDTO taskDTO, CompletableFuture<TaskInvokeRes> future) {
        this.taskDTO = taskDTO;
        this.future = future;
    }

    @Override
    public void run() {
        TaskType taskType = taskDTO.getTaskType();
        AutomaticTaskRunner taskRunner = new AutomaticTaskRunner() {
            final AutomaticTaskRunner proxy = JSONObject.parseObject(taskDTO.getTaskConfig(), AutomaticTaskRunner.class);

            @Override
            public TaskType taskType() {
                return proxy.taskType();
            }

            @Override
            public boolean doTask() {
                log.info("before dotask {} currtime {}, invokeTime {}",
                        taskDTO.getBizId(),
                        DateUtil.formatDateTime(new Date()),
                        DateUtil.formatDateTime(taskDTO.getNextInvokeTime())
                );
                try {
                    return proxy.doTask();
                } catch (Exception e) {
                    log.error("", e);
                    throw new RuntimeException(e);
                }
            }
        };
        Date now = new Date();
        AssertUtil.isTrue(taskRunner.taskType() == taskType);
        try {
            int retryNum = 0;
            boolean invokeSuccess = false;
            Exception error = null;
            while (!invokeSuccess && retryNum <= taskDTO.getRetryCount()) {
                try {
                    if (taskDTO.getTimeLimit() != null && taskDTO.getTimeLimit() > 0) {
                        invokeSuccess = CompletableFuture.supplyAsync(taskRunner::doTask)
                                .get(taskDTO.getTimeLimit(), TimeUnit.MILLISECONDS);
                    } else {
                        invokeSuccess = CompletableFuture.supplyAsync(taskRunner::doTask).get();
                    }
                } catch (Exception e) {
                    log.error("", e);
                    error = e;
                } finally {
                    retryNum++;
                }
            }
            future.complete(TaskInvokeRes.builder().taskStartTime(now).taskEndTime(new Date())
                    .taskInvokerNum(retryNum).invokerRes(invokeSuccess).error(error).build());
        } catch (Exception e) {
            log.error("", e);
            future.complete(TaskInvokeRes.builder().taskStartTime(now)
                    .taskEndTime(new Date()).error(e).build());
        }

    }
}

